package com.yunhe.transfer.controller.data;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhe.common.constant.KafkaConsts;
import com.yunhe.common.model.ResultObject;
import com.yunhe.common.model.model.RunStrategyDistributeDTO;
import com.yunhe.transfer.service.data.RunStrategyService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;

import java.util.Map;
import java.util.Optional;

@RestController
@RequestMapping("/runStrategy")
public class RunStrategyController {

    @Autowired
    private RunStrategyService runStrategyService;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    private static final ObjectMapper mapper = new ObjectMapper();

    private static final Logger logger = LoggerFactory.getLogger(RunStrategyController.class);

    /**
     * 下发策略
     * @param runStrategyDistributeDTO
     * @param needKafka
     * @param kafkaKey
     */
    @RequestMapping(value = "/remote/distribute", method = RequestMethod.POST)
    public ResponseEntity distributeRunStrategy(@RequestBody RunStrategyDistributeDTO runStrategyDistributeDTO,
                                                @RequestParam("needKafka") Optional<Boolean> needKafka,
                                                @RequestParam("kafkaKey") Optional<String> kafkaKey) throws Exception {
        ResultObject result = runStrategyService.distributeRunStrategy(runStrategyDistributeDTO);
        if (needKafka.isPresent() && needKafka.get()) {
            kafkaTemplate.send(KafkaConsts.Topic.RunStrategyDistributeResult, kafkaKey.orElse(""), mapper.writeValueAsString(result));
        }
        return new ResponseEntity<>(result, HttpStatus.OK);
    }

    /**
     * Kafka监听策略下发消息
     * @param record
     */
    @KafkaListener(topics = KafkaConsts.Topic.RunStrategyDistribute, groupId = KafkaConsts.Group.RandomGroupId)
    public void distributeRunStrategy(ConsumerRecord<String, String> record) {
        try {
            String key = record.key();
            String value = record.value();
            RunStrategyDistributeDTO runStrategyDistributeDTO = mapper.readValue(value, RunStrategyDistributeDTO.class);
            distributeRunStrategy(runStrategyDistributeDTO, Optional.of(true), Optional.ofNullable(key));
        } catch (Exception e) {
            logger.error("", e);
        }
    }

    /**
     * 获取本地控制参数
     */
    @RequestMapping(value = "/local", method = RequestMethod.POST)
    public ResponseEntity getLocalRunStrategy(@RequestBody RunStrategyDistributeDTO runStrategyDistributeDTO,
                                              @RequestParam("needKafka") Optional<Boolean> needKafka,
                                              @RequestParam("kafkaKey") Optional<String> kafkaKey) throws Exception {
        ResultObject result = runStrategyService.getLocalRunStrategy(runStrategyDistributeDTO);
        if (needKafka.isPresent() && needKafka.get()) {
            kafkaTemplate.send(KafkaConsts.Topic.LocalRunStrategyResult, kafkaKey.orElse(""), mapper.writeValueAsString(result));
        }
        return new ResponseEntity<>(result, HttpStatus.OK);
    }

    /**
     * Kafka监听获取本地控制参数消息
     * @param record
     */
    @KafkaListener(topics = KafkaConsts.Topic.LocalRunStrategy, groupId = KafkaConsts.Group.RandomGroupId)
    public void getLocalRunStrategy(ConsumerRecord<String, String> record) {
        try {
            String key = record.key();
            String value = record.value();
            RunStrategyDistributeDTO runStrategyDistributeDTO = mapper.readValue(value, RunStrategyDistributeDTO.class);
            if (runStrategyDistributeDTO.getStationCode() != null) {
                getLocalRunStrategy(runStrategyDistributeDTO, Optional.of(true), Optional.ofNullable(key));
            }
        } catch (Exception e) {
            logger.error("", e);
        }
    }

    /**
     * 下发节假日信息
     * @param holidays
     */
    @RequestMapping(value = "/holiday", method = RequestMethod.POST)
    public ResponseEntity distributeHolidays(@RequestBody Map<String, Object> holidays) throws Exception {
        ResultObject result = runStrategyService.distributeHolidays(holidays);
        return new ResponseEntity<>(result, HttpStatus.OK);
    }
}
